package com.advert.mqtt.client;

import com.alibaba.fastjson.JSONObject;
import com.advert.mqtt.enums.QosEnum;
import com.advert.mqtt.properties.MqttProperties;
import com.advert.service.IDeviceService;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * Created by 传智播客*黑马程序员.
 */
@Component
public class EmqClient2 {

    private static final Logger log = LoggerFactory.getLogger(EmqClient2.class);

    private IMqttAsyncClient mqttClient;

    @Autowired
    private MqttProperties mqttProperties;


    @Value("${mqtt.clinet-id2}")
    String clientId;

    @Autowired
    EmqClient2 client;

    @Autowired
    private IDeviceService deviceService;

    @Value("${cabinet.cabinetTopic}")
    String cabinetTopic;
    @Value("${cabinet.statusTopic}")
    String statusTopic;

    public boolean isConnected() {
        if (mqttClient != null) {
            return mqttClient.isConnected();
        }
        return false;
    }


    @PostConstruct
    public void init() {
        MqttClientPersistence mempersitence = new MemoryPersistence();
        try {
            mqttClient = new MqttAsyncClient(mqttProperties.getBrokerUrl(), clientId, mempersitence);
        } catch (MqttException e) {
            log.error("初始化客户端mqttClient对象失败,errormsg={},brokerUrl={},clientId={}", e.getMessage(), mqttProperties.getBrokerUrl(), mqttProperties.getClientId());
        }
    }

    /**
     * 连接broker
     *
     * @param username
     * @param password
     */
    public void connect(String username, String password) {
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setAutomaticReconnect(true);
        options.setUserName(username);
        options.setPassword(password.getBytes());
        options.setCleanStart(false);
        mqttClient.setCallback(new MessageCallback());
        log.info("mqtt 初始化连接...");
        try {
            mqttClient.connect(options, null, new MqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.info("mqtt connect success");
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
                    throwable.printStackTrace();
                    log.info("mqtt 连接失败 {}", throwable.getMessage());
                }
            });
        } catch (MqttException e) {
            log.error("mqtt客户端连接服务端失败,失败原因{}", e.getMessage());
        }
    }

    /**
     * 断开连接
     */
    @PreDestroy
    public void disConnect() {
        try {
            mqttClient.disconnect();
        } catch (MqttException e) {
            log.error("断开连接产生异常,异常信息{}", e.getMessage());
        }
    }

    /**
     * 重连
     */
    public void reConnect() {
        try {
            mqttClient.reconnect();
        } catch (MqttException e) {
            log.error("重连失败,失败原因{}", e.getMessage());
        }
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param msg
     * @param qos
     * @param retain
     */
    public void publish(String topic, String msg, QosEnum qos, boolean retain) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        try {
            mqttClient.publish(topic, mqttMessage, null, new MqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.info("发布消息成功:topic:{} message:{}", topic, mqttMessage);
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.error("发布消息失败,errormsg={},topic={},msg={},error={}", e.getMessage(), topic, msg, e.getMessage());
                    e.printStackTrace();
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param msg
     */
    public void publish(String topic, String msg) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        mqttMessage.setQos(QosEnum.QoS1.value());
        try {
            log.info("发布消息成功:topic:{} message:{}", topic, mqttMessage);
            mqttClient.publish(topic, mqttMessage);
        } catch (MqttException e) {
            log.error("发布消息失败,errormsg={},topic={},msg={},qos={},retain={}", e.getMessage(), topic, msg, 0, false);
        }
    }

    /**
     * 订阅
     *
     * @param topicFilter
     * @param qos
     */
    public void subscribe(String topicFilter, QosEnum qos) {
        try {
            log.info("topicFilter====" + topicFilter);
            mqttClient.subscribe(topicFilter, qos.value());
        } catch (MqttException e) {
            log.error("订阅主题失败,errormsg={},topicFilter={},qos={}", e.getMessage(), topicFilter, qos.value());
        }

    }

    /**
     * 取消订阅
     *
     * @param topicFilter
     */
    public void unSubscribe(String topicFilter) {
        try {
            mqttClient.unsubscribe(topicFilter);
        } catch (MqttException e) {
            log.error("取消订阅失败,errormsg={},topicfiler={}", e.getMessage(), topicFilter);
        }
    }

    class MessageCallback implements MqttCallback {

        @Override
        public void deliveryComplete(IMqttToken token) {
            int messageId = token.getMessageId();
            String[] topics = token.getTopics();
            log.info("消息发布完成,messageid={},topics={}", messageId, topics);
        }


        @Override
        public void disconnected(MqttDisconnectResponse mqttDisconnectResponse) {
            // 资源的清理 重连
            log.info("丢失了对服务端的连接,重新连接");
            client.reConnect();
        }

        @Override
        public void mqttErrorOccurred(MqttException e) {
            e.printStackTrace();
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) {
            try {
                log.info("{} 收到mqtt消息messageArrived:{}", clientId, message.toString());
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("MqttBody", message.getPayload());
                jsonObject.put("MqttTopic", topic);

                deviceService.handlerCabinet(jsonObject);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void connectComplete(boolean reconnect, String s) {
            log.info("mqtt 连接成功 订阅主题");

            log.info("cabinetTopic====" + cabinetTopic);
            log.info("statusTopic====" + statusTopic);
            // 订阅一个主题
            client.subscribe(cabinetTopic, QosEnum.QoS1);
            // 订阅在线状态主题
            if (!StringUtils.isEmpty(statusTopic)) {
                client.subscribe(statusTopic, QosEnum.QoS0);
            }
        }

        @Override
        public void authPacketArrived(int i, org.eclipse.paho.mqttv5.common.packet.MqttProperties mqttProperties) {

        }
    }
}
